package com.apps.sdses.flink141.demo;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;

/**
 * @author jiwei
 * @description
 * @date 2023/11/6 16:37
 */
public class CSV2DorisDemo {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //设置时区
        tableEnv.getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai"));

        //Source
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    s_id STRING,\n" +
                        "\tdt TIMESTAMP(3),\n" +
                        "\ts_name STRING,\n" +
                        "\tserial_number int,\n" +
                        "\ts_code STRING,\n" +
                        "\tsaving_grade STRING,\n" +
                        "\tbuild_area decimal(10, 2),\n" +
                        "\theating_area decimal(10, 2),\n" +
                        "\tflow_upper decimal(10, 1),\n" +
                        "\tflow_lower decimal(10, 1),\n" +
                        "\tt_supply_upper decimal(10, 1),\n" +
                        "\tt_supply_lower decimal(10, 1),\n" +
                        "\ttopology_distance decimal(10, 2),\n" +
                        "\trysl int,\n" +
                        "\tjzsl int,\n" +
                        "\tygwd decimal(10, 1),\n" +
                        "\tyhwd decimal(10, 1),\n" +
                        "\tygyl decimal(10, 2),\n" +
                        "\tyhyl decimal(10, 2),\n" +
                        "\tssll decimal(10, 1),\n" +
                        "\tssrl decimal(10, 1),\n" +
                        "\tljll decimal(10, 1),\n" +
                        "\tljrl decimal(10, 1),\n" +
                        "\trbgw decimal(10, 1),\n" +
                        "\trbhw decimal(10, 1),\n" +
                        "\tjz1_egwd decimal(10, 1),\n" +
                        "\tjz1_ehwd decimal(10, 1),\n" +
                        "\tjz1_egyl decimal(10, 2),\n" +
                        "\tjz1_ehyl decimal(10, 2),\n" +
                        "\tjz2_egwd decimal(10, 1),\n" +
                        "\tjz2_ehwd decimal(10, 1),\n" +
                        "\tjz2_egyl decimal(10, 2),\n" +
                        "\tjz2_ehyl decimal(10, 2),\n" +
                        "\tjz3_egwd decimal(10, 1),\n" +
                        "\tjz3_ehwd decimal(10, 1),\n" +
                        "\tjz3_egyl decimal(10, 2),\n" +
                        "\tjz3_ehyl decimal(10, 2),\n" +
                        "\tjz1_ddtjfkd decimal(10, 2),\n" +
                        "\tjz2_ddtjfkd decimal(10, 2),\n" +
                        "\tjz3_ddtjfkd decimal(10, 2),\n" +
                        "\tswwd decimal(10, 1),\n" +
                        "\tjz1_rbgw decimal(10, 1),\n" +
                        "\tjz1_rbhw decimal(10, 1),\n" +
                        "\tjz1_ssll decimal(10, 1),\n" +
                        "\tjz1_ssrl decimal(10, 1),\n" +
                        "\tjz1_ljll decimal(10, 1),\n" +
                        "\tjz1_ljrl decimal(10, 1),\n" +
                        "\tjz1_ljbsl decimal(10, 1),\n" +
                        "\tjz2_rbgw decimal(10, 1),\n" +
                        "\tjz2_rbhw decimal(10, 1),\n" +
                        "\tjz2_ssll decimal(10, 1),\n" +
                        "\tjz2_ssrl decimal(10, 1),\n" +
                        "\tjz2_ljll decimal(10, 1),\n" +
                        "\tjz2_ljrl decimal(10, 1),\n" +
                        "\tjz2_ljbsl decimal(10, 1),\n" +
                        "\tjz3_rbgw decimal(10, 1),\n" +
                        "\tjz3_rbhw decimal(10, 1),\n" +
                        "\tjz3_ssll decimal(10, 1),\n" +
                        "\tjz3_ssrl decimal(10, 1),\n" +
                        "\tjz3_ljll decimal(10, 1),\n" +
                        "\tjz3_ljrl decimal(10, 1),\n" +
                        "\tjz3_ljbsl decimal(10, 1),\n" +
                        "\tdrwqyl decimal(10, 2),\n" +
                        "\tdrwhyl decimal(10, 2),\n" +
                        "\tdrhsyl decimal(10, 2),\n" +
                        "\tdrrbssll decimal(10, 1),\n" +
                        "\tdrrbssrl decimal(10, 1),\n" +
                        "\tdrrbljll decimal(10, 1),\n" +
                        "\tdrrbljrl decimal(10, 1),\n" +
                        "\tdrrbgw decimal(10, 1),\n" +
                        "\tdrrbhw decimal(10, 1),\n" +
                        "\tygdd decimal(10, 1)" +
                "     ) WITH (\n" +
                "     'connector' = 'filesystem',\n" +
                "     'path' = 'E:\\workspace\\ss-flink\\src\\main\\resources\\data\\SHRL_HRZ_FORMAL_202311061532.csv',\n" +
                "     'format' = 'csv' " +
//                "     'field-delimiter' = '|'" +
                ")");

        //Sink
//        tableEnv.executeSql("CREATE TABLE sink_people_info (\n" +
//                "     id INT,\n" +
//                "     hobby STRING,\n" +
//                "     school STRING,\n" +
//                "     PRIMARY KEY(id) NOT ENFORCED\n" +
//                "     ) WITH (\n" +
//                "     'connector' = 'upsert-kafka',\n" +
//                "     'topic' = 'yzq_test',\n" +
//                "     'properties.bootstrap.servers' = '192.168.102.154:29092',\n" +
//                "     'key.format' = 'json',\n" +
//                "     'value.format' = 'json')");

        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
                "LIKE sourceTable (EXCLUDING ALL)");

        //Insert
        tableEnv.executeSql("insert into sinkTable\n" +
                "SELECT * FROM sourceTable limit 100");

    }
}
